package com.bauer.java.hadoop.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Iterator;

/**
 * @Author ：rui.wei
 * @Date ：Created in 17:02 2019/8/12
 * @Description：
 */
public class HbaseCient {

    public static String TABLE_NAME = "sales";

    public static void main(String[] args) throws Exception {
//        createSchemaTables(initHbase());
        createSchemaTables(initHbase());
    }

    public static void queryData(Connection connection) throws Exception {
        HTable table = (HTable) connection.getTable(TableName.valueOf("sales1"));
        Scan scan = new Scan();
        ResultScanner resultScanner = table.getScanner(scan);
        Iterator<Result> iterator = resultScanner.iterator();
        while (iterator.hasNext()) {
            Result result = iterator.next();

            for (Cell cell : result.listCells()) {
                System.out
                        .println("row:" + Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
                System.out.println("family:"
                        + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));

                System.out.println("qualifier:"
                        + Bytes
                        .toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));

                String qualifier = Bytes
                        .toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                if (qualifier.equals("year")) {
                    System.out
                            .println("value:" + Bytes
                                    .toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                } else {
                    System.out
                            .println("value:" + Bytes
                                    .toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                }

                System.out.println("timestamp:" + cell.getTimestamp());
                System.out
                        .println("-------------------------------------------");
            }
        }
        resultScanner.close();
        connection.close();
    }

    public static void insertData(Connection connection) throws Exception {
        HTable table = (HTable) connection.getTable(TableName.valueOf("sales1"));
        Put put = new Put(Bytes.toBytes("ceshi1"));
        put.addColumn(Bytes.toBytes("baseinfo"), Bytes.toBytes("city"), Bytes.toBytes("南京"));
        table.put(put);
        connection.close();
    }

    public static void createSchemaTables(Connection connection) throws IOException {
        String TABLE_NAME = "sales1";
        String[] column_families = new String[]{"baseinfo", "sales"};
        Admin admin = connection.getAdmin();

        HTableDescriptor table = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
        for (String cf : column_families) {
            table.addFamily(new HColumnDescriptor(cf));
        }

        byte[][] splitKeys = {"1002022".getBytes()};
        if (admin.tableExists(TableName.valueOf(TABLE_NAME))) {
            System.out.println("table Exists!");
        } else {
            admin.createTable(table, splitKeys);
            System.out.println("create table Success!");
        }
        connection.close();
    }

    public static void dropTables(Connection connection) throws IOException {
        String TABLE_NAME = "sales1";
        Admin admin = connection.getAdmin();
        admin.disableTable(TableName.valueOf(TABLE_NAME));
        admin.deleteTable(TableName.valueOf(TABLE_NAME));

        connection.close();
    }

    public static void getSchemaTables(Connection connection) throws IOException {
        String TABLE_NAME = "sales";
        Admin admin = connection.getAdmin();
        HTableDescriptor table = admin.getTableDescriptor(TableName.valueOf(TABLE_NAME));
        HColumnDescriptor[] columnDescriptors = table.getColumnFamilies();
        for (HColumnDescriptor columnDescriptor : columnDescriptors) {
            System.out.println("family:"
                    + Bytes.toString(columnDescriptor.getName()));
        }
        connection.close();
    }

    public static void deleteData(Connection connection) throws IOException {
        HTable table = (HTable) connection.getTable(TableName.valueOf("sales1"));
        Scan scan = new Scan();
        ResultScanner resultScanner = table.getScanner(scan);
        Iterator<Result> iterator = resultScanner.iterator();
        while (iterator.hasNext()) {
            Result result = iterator.next();

            table.delete(new Delete(result.getRow()));
        }
        connection.close();
    }


    public static Connection initHbase() throws IOException {
        System.setProperty("hadoop.home.dir", "E:\\Program Files\\hadoop-common-2.2.0-bin");
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.zookeeper.quorum", "99.48.58.128");
        //集群配置↓
        //configuration.set("hbase.zookeeper.quorum", "101.236.39.141,101.236.46.114,101.236.46.113");
        Connection connection = ConnectionFactory.createConnection(configuration);
        return connection;
    }
}
